package com.zhaosc.rx.example;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.subscribers.DisposableSubscriber;

public class HelloWorld {


	@Test
	public void test02() {
		Observable<String> observable1 = Observable.just("a", "b", "c");
		List<Integer> list = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8));
		Observable<Integer> observable2 = Observable.fromIterable(list);
		observable2.subscribe(item -> System.out.println(item), error -> error.printStackTrace(),
				() -> System.out.println("Done"));
	}


	@Test
	public void testFromFuture() {
		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

		Future future = executor.scheduleAtFixedRate(() -> {
			System.out.println("Hello world!");
		}, 1L, 1L, TimeUnit.SECONDS);

		Observable<String> observable = Observable.fromFuture(future);

		observable.subscribe(item -> System.out.println(item), error -> error.printStackTrace(),
				() -> System.out.println("Done"));

		executor.shutdown();
	}

	@Test
	public void testGenerate() {
		int startValue = 1;
		int incrementValue = 1;
		Flowable<Integer> flowable = Flowable.generate(() -> startValue, (s, emitter) -> {
			int nextValue = s + incrementValue;
			emitter.onNext(nextValue);
			if (nextValue == 100) {
				emitter.onError(new RuntimeException("limited"));
			}
			return nextValue;
		});
		flowable.subscribe(value -> System.out.println(value));
	}

	@Test
	public void testCreate() {
		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

		ObservableOnSubscribe<String> handler = emitter -> {

			Future<Object> future = executor.schedule(() -> {
				emitter.onNext("Hello");
				emitter.onNext("World");
				emitter.onComplete();
				return null;
			}, 1, TimeUnit.SECONDS);

			emitter.setCancellable(() -> future.cancel(false));
		};

		Observable<String> observable = Observable.create(handler);

		observable.subscribe(item -> System.out.println(item), error -> error.printStackTrace(),
				() -> System.out.println("Done"));

		executor.shutdown();
	}

	@Test
	public void testInterval() {
		Observable<Long> clock = Observable.interval(1, TimeUnit.SECONDS);

		clock.subscribe(time -> {
			if (time % 2 == 0) {
				System.out.println("Tick");
			} else {
				System.out.println("Tock");
			}
		});
	}
	
	@Test
	public void testTimer() {
		Observable<Long> eggTimer = Observable.timer(5, TimeUnit.SECONDS);

		eggTimer.blockingSubscribe(v -> System.out.println("Egg is ready!"));
	}

}
